package edu.buaa.act.model.tsc;

import edu.buaa.act.anonymous.TrajectorySeparator;
import edu.buaa.act.helper.Utils;
import edu.buaa.act.model.Trajectory;
import org.slf4j.Logger;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import static edu.buaa.act.helper.Utils.percent;
/**
 * Created by song on 17-7-5.
 */
public class MapMatchingServer
{
    private static Logger log = Utils.getDebugLogger();
    private final int parallelCount;
    private final MyMapMatching matching;
    private final boolean breakLongTraj;

    private BlockingQueue<Trajectory> trajBeforeMapMatching = new LinkedBlockingQueue<>();
    private BlockingQueue<Trajectory> trajAfterMapMatching = new LinkedBlockingQueue<>();
    private List<MatchingThread> threadList = new ArrayList<>();
    private AtomicInteger currentRunning = new AtomicInteger(0);
    private AtomicLong taskCompleteCount = new AtomicLong(0);
    private AtomicLong outputCount = new AtomicLong(0);
    private AtomicLong waitCount = new AtomicLong(0);
    private AtomicLong err_tooShort = new AtomicLong(0);
    private AtomicLong err_empty = new AtomicLong(0);
    private AtomicLong err_travelTime = new AtomicLong(0);
    private AtomicLong err_lostWay = new AtomicLong(0);
    private AtomicLong err_tooLong = new AtomicLong(0);
    private AtomicLong err_other = new AtomicLong(0);
    private Map<String, List<Trajectory>> matchFailedTrajs = new HashMap<>();

    public MapMatchingServer(int parallelCount, File map, int gpsAccuracy, boolean breakTooLongTraj) throws IOException {
        this.matching = new MyMapMatching(map, Utils.getOrCreateDir("map-matching-cache"), gpsAccuracy);
        this.parallelCount = parallelCount;
        this.breakLongTraj = breakTooLongTraj;
    }

    public void start()
    {
        for(int i=0; i<parallelCount; i++)
        {
            MatchingThread thread = new MatchingThread(i);
            threadList.add(thread);
            thread.start();
        }

    }


    public void put(Trajectory traj) throws InterruptedException {
        this.trajBeforeMapMatching.put(traj);
    }

    public Trajectory poll(int i, TimeUnit seconds) throws InterruptedException {
        return this.trajAfterMapMatching.poll(i, seconds);
    }

    public void showLog(){
        long in = taskCompleteCount.get();
        long out = outputCount.get();
        long diff = in-out;
        if(in==0) {log.trace("NO OUTPUT! ALL TRAJ MapMatching FAILED!");return;}
        log.trace("map-matching statistics: out({})/in({})={}% wait({} seconds)", out, in, out*100/in, waitCount.get());
        log.trace("map-matching statistics: in({})-out({})={}(total) [lost {}%, empty {}%, travelTime {}%, tooLong {}%, tooShort {}%, other {}%]",
                in, out, diff,
                percent(err_lostWay.get(), diff),
                percent(err_empty.get(), diff),
                percent(err_travelTime.get(), diff),
                percent(err_tooLong.get(), diff),
                percent(err_tooShort.get(),diff),
                percent(err_other.get(), diff));
    }

    public float getInOutRate(){
        return outputCount.get()*1f/taskCompleteCount.get();
    }

    public void setOutput(BlockingQueue<Trajectory> output)
    {
        this.trajAfterMapMatching = output;
    }

    public void join(){
        for (MatchingThread t : threadList)
        {
            try {
                t.join();
                for(Map.Entry<String, List<Trajectory>> entry : t.matchFailedTrajs.entrySet()){
                    this.matchFailedTrajs.putIfAbsent(entry.getKey(), new ArrayList<>());
                    this.matchFailedTrajs.get(entry.getKey()).addAll(entry.getValue());
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public boolean done() {
        return (this.trajBeforeMapMatching.isEmpty() && this.currentRunning.get()<=0);
    }

    public void printStatus()
    {
        log.trace("IN WAIT({}) OUT WAIT({})", trajBeforeMapMatching.size(), trajAfterMapMatching.size());
    }

    public Map<String, List<Trajectory>> getFailedTrajs() {
        return matchFailedTrajs;
    }

    private class MatchingThread extends Thread {
        private final int threadId;
        private Map<String, List<Trajectory>> matchFailedTrajs = new HashMap<>();

        public MatchingThread(int i) {
            this.threadId = i;
        }

        public void run() {
            Thread.currentThread().setName(String.format("MM%02d", threadId));
            currentRunning.getAndIncrement();
            try {
                while (true) {
                    Trajectory traj = trajBeforeMapMatching.poll(5, TimeUnit.SECONDS);
                    if (traj == null) {
                        break;
                    } else {
                        if (traj.getGpsEntries().size() > 1) {
                            try {
                                matching.matching(traj);
                                trajAfterMapMatching.put(traj);
                                outputCount.getAndIncrement();

                            } catch (IllegalStateException e) {
                                err_other.getAndIncrement();
                                addToFailedList("OTHER", traj);
                            } catch (MyMapMatching.NoEdgeMatchedException e) {
                                err_empty.getAndIncrement();
                                addToFailedList("NO_ROAD", traj);
                            } catch (MyMapMatching.CannotCalcTravelTimeException e) {
                                err_travelTime.getAndIncrement();
                                addToFailedList("NO_TIME", traj);
                            } catch (IllegalArgumentException e) {
                                err_lostWay.getAndIncrement();
                                addToFailedList("GET_LOST", traj);
                            } catch (RuntimeException e) {

                                if (e.getMessage() != null && e.getMessage().startsWith("Sequence is broken for submitted track at time step")) {
//                                            long name = System.currentTimeMillis()/1000;
//                                            traj.toFile(Utils.getFile("data/error/broken-"+traj.isSample()+name+".gpx"));
//                                            errLog.append(name).append('\t').append(e.getMessage()).append('\n');

                                    if (breakLongTraj) {
                                        for (Trajectory t : TrajectorySeparator.separateTrajTooLong(traj)) {
                                            trajBeforeMapMatching.put(t);
                                            taskCompleteCount.getAndIncrement();
                                        }
                                    } else {
                                        err_tooLong.getAndIncrement();
                                        addToFailedList("TOO_LONG", traj);
                                    }
                                } else {
                                    addToFailedList("UNKNOWN", traj);
                                }
                            }
                        } else {
                            err_tooShort.getAndIncrement();
                            addToFailedList("TOO_SHORT", traj);
                        }
                        taskCompleteCount.getAndIncrement();
                    }
                }
            } catch (InterruptedException e) {
                System.out.print("map-matching thread interrupted.");
            } catch (Throwable e) {
                log.error("Error in MM thread!", e);
            } finally {
                currentRunning.getAndDecrement();
            }
        }

        private void addToFailedList(String reason, Trajectory trajectory) {
            matchFailedTrajs.computeIfAbsent(reason, k -> new ArrayList<>());
            matchFailedTrajs.get(reason).add(trajectory);
        }
    }

}
